package com.fulihui.wgj.biz.impl;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.fulihui.common.enums.TemplateLogTypeEnum;
import com.fulihui.common.util.JedisUtil;
import com.fulihui.weixinsdk.model.TemplateItem;
import com.fulihui.wgj.biz.AutoTaskBiz;
import com.fulihui.wgj.biz.WxUserInfoBiz;
import com.fulihui.wgj.biz.common.util.TemplateRenderUtil;
import com.fulihui.wgj.dal.impala.domain.WxUserInfoSend;
import com.fulihui.wgj.dal.mybatis.domain.TemplateAutoLog;
import com.fulihui.wgj.dal.mybatis.domain.TemplateAutoTask;
import com.fulihui.wgj.dal.mybatis.domain.WxTemplate;
import com.fulihui.wgj.dal.mybatis.repository.TemplateAutoTaskRepository;
import com.fulihui.wgj.dal.mybatis.repository.WxTemplateRepository;
import com.fulihui.wgj.facade.dto.MsgUserDTO;
import com.fulihui.wgj.facade.dto.TemplateMsgDTO;
import com.fulihui.wgj.facade.msg.TemplateMsg;
import com.fulihui.wgj.facade.request.autotask.AutoTaskSaveRequest;
import com.fulihui.wgj.facade.request.autotask.CalcAutoCountRequest;
import com.fulihui.wgj.facade.request.template.CalcSendCountRequest;
import com.fulihui.wgj.mq.kafka.producer.KafkaProducer;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 自动化任务业务
 *
 * @author wang_wx
 * @date 2018-08-02
 */
@Service
public class AutoTaskBizImpl implements AutoTaskBiz {

    private static final Logger LOGGER = LoggerFactory.getLogger(AutoTaskBizImpl.class);

    /**
     * 自动任务发送到kafka队列消息redis_key
     */
    private static final String AUTO_TASK_SEND_REDIS_PREFIX = "AUTO_TASK_SEND_";

    /**
     * 队列批次长度--一般
     */
    private static final int MQ_LENGTH_GENERALLY = 200;

    @Autowired
    TemplateAutoTaskRepository templateAutoTaskRepository;

    @Autowired
    WxTemplateRepository wxTemplateRepository;

    @Autowired
    ThreadPoolTaskExecutor threadPool;

    @Autowired
    WxUserInfoBiz wxUserInfoBiz;

    @Autowired
    JedisUtil jedisUtil;

    @Autowired
    KafkaProducer kafkaProducer;

    @Override
    public CalcSendCountRequest trance2CalcSendRequest(CalcAutoCountRequest request) {
        if (Objects.nonNull(request)) {
            CalcSendCountRequest calcSendCountRequest = new CalcSendCountRequest();
            calcSendCountRequest.setAppId(request.getAppId());
            calcSendCountRequest.setCountry(request.getCountry());
            calcSendCountRequest.setCountryTab(request.getCountryTab());
            calcSendCountRequest.setProvince(request.getProvince());
            calcSendCountRequest.setProvinceTab(request.getProvinceTab());
            calcSendCountRequest.setCity(request.getCity());
            calcSendCountRequest.setSex(request.getSex());
            calcSendCountRequest.setSexTab(request.getSexTab());
            calcSendCountRequest.setVip(request.getVip());
            calcSendCountRequest.setOs(request.getOs());
            calcSendCountRequest.setOsTab(request.getOsTab());
            calcSendCountRequest.setHasClickLink(request.getHasClickLink());
            calcSendCountRequest.setHasClickLinkTab(request.getHasClickLinkTab());
            calcSendCountRequest.setHasClickMiniProgram(request.getHasClickMiniProgram());
            calcSendCountRequest.setHasClickMiniProgramTab(request.getHasClickMiniProgramTab());
            calcSendCountRequest.setYzFilter(request.getYzFilter());
            calcSendCountRequest.setContainRequest(request.getContainRequest());
            calcSendCountRequest.setEliminateRequest(request.getEliminateRequest());
            return calcSendCountRequest;
        }
        return null;
    }

    @Override
    public Long saveTaskByRequest(AutoTaskSaveRequest request) {
        CalcAutoCountRequest autoCountRequest = request.getCalcAutoCountRequest();
        TemplateAutoTask autoTask = new TemplateAutoTask();
        autoTask.setAppId(request.getAppId());
        autoTask.setTaskName(request.getTaskName());
        autoTask.setStatus(request.getStatus());
        autoTask.setCycle(request.getCycle());
        autoTask.setSendTime(request.getSendTime());
        autoTask.setJumpUrl(request.getTemplateUrl());
        autoTask.setLogType(autoCountRequest.getLogType());
        autoTask.setMiniApp(request.getMiniApp());
        autoTask.setMiniAppId(request.getMiniAppId());
        autoTask.setMiniProgramUrl(request.getMiniAppUrl());
        autoTask.setSendCondition(JSON.toJSONString(autoCountRequest));
        autoTask.setSendConditionDesc(request.getSendConditionDesc());
        autoTask.setTemplateId(request.getTemplateId());
        String templateContent = request.getTemplateContent();
        // 模板发送数据
        Map<String, TemplateItem> templateData = JSON.parseObject(templateContent,
                new TypeReference<Map<String, TemplateItem>>() {
                });
        autoTask.setTemplateContent(renderTemplate(templateData, request.getTemplateId()));
        autoTask.setTemplateData(templateContent);
        templateAutoTaskRepository.saveSelective(autoTask);
        LOGGER.info("saveTaskByRequest task id = ", autoTask.getId());
        return autoTask.getId();
    }

    @Override
    public Integer updateTaskByRequest(AutoTaskSaveRequest request) {
        CalcAutoCountRequest autoCountRequest = request.getCalcAutoCountRequest();
        TemplateAutoTask autoTask = new TemplateAutoTask();
        autoTask.setId(request.getId());
        autoTask.setAppId(request.getAppId());
        autoTask.setTaskName(request.getTaskName());
        autoTask.setStatus(request.getStatus());
        autoTask.setCycle(request.getCycle());
        autoTask.setSendTime(request.getSendTime());
        autoTask.setJumpUrl(request.getTemplateUrl());
        autoTask.setLogType(autoCountRequest.getLogType());
        autoTask.setMiniApp(request.getMiniApp());
        autoTask.setMiniAppId(request.getMiniAppId());
        autoTask.setMiniProgramUrl(request.getMiniAppUrl());
        autoTask.setSendCondition(JSON.toJSONString(autoCountRequest));
        autoTask.setSendConditionDesc(request.getSendConditionDesc());
        autoTask.setTemplateId(request.getTemplateId());
        String templateContent = request.getTemplateContent();
        // 模板发送数据
        Map<String, TemplateItem> templateData = JSON.parseObject(templateContent,
                new TypeReference<Map<String, TemplateItem>>() {
                });
        autoTask.setTemplateContent(renderTemplate(templateData, request.getTemplateId()));
        autoTask.setTemplateData(templateContent);
        Integer count = templateAutoTaskRepository.updateSelective(autoTask);
        LOGGER.info("updateTaskByRequest count = ", count);
        return count;
    }

    /**
     * 渲染模板内容
     *
     * @param data
     * @param templateId 模板ID
     * @return
     */
    private String renderTemplate(Map<String, TemplateItem> data, String templateId) {
        WxTemplate wxTemplateQuery = new WxTemplate();
        wxTemplateQuery.setTemplateId(templateId);
        List<WxTemplate> templateList = wxTemplateRepository.queryByWhere(wxTemplateQuery);
        if (!CollectionUtils.isEmpty(templateList)) {
            WxTemplate wxTemplate = templateList.get(0);
            if (Objects.nonNull(wxTemplate)) {
                return TemplateRenderUtil.render(data, wxTemplate);
            }
        }
        return null;
    }

    @Override
    public void sendAutoMsg(CalcSendCountRequest request, TemplateAutoTask templateAutoTask, TemplateAutoLog templateAutoLog) {
        LOGGER.info("sendAutoMsg start .......");
        Integer logType = templateAutoTask.getLogType();
        LOGGER.info("logType = {}", logType);
        Long recordCount = templateAutoLog.getSendCount();
        if (recordCount == null || recordCount == 0L) {
            if (TemplateLogTypeEnum.SHOP.getValue() == logType.intValue()) {
                recordCount = wxUserInfoBiz.queryCount(request);
            } else if (TemplateLogTypeEnum.GAME.getValue() == logType.intValue()) {
                recordCount = wxUserInfoBiz.queryCount4Game(request);
            }
        }
        if (recordCount == 0L) {
            LOGGER.info("recordCount = 0 , do not send !");
            return;
        }

        // 发送的模板数据
        StringBuilder sendRedisKey = new StringBuilder(AUTO_TASK_SEND_REDIS_PREFIX);
        sendRedisKey.append(request.getAppId()).append("_").append(templateAutoLog.getId());

        TemplateMsgDTO templateMsgDTO = new TemplateMsgDTO();
        templateMsgDTO.setTemplateId(templateAutoTask.getTemplateId());
        templateMsgDTO.setTemplateUrl(templateAutoTask.getJumpUrl());
        templateMsgDTO.setMiniProgramAppId(templateAutoTask.getMiniAppId());
        templateMsgDTO.setMiniProgramUrl(templateAutoTask.getMiniProgramUrl());
        // 模板发送数据
        String templateContent = templateAutoTask.getTemplateData();
        Map<String, TemplateItem> data = JSON.parseObject(templateContent,
                new TypeReference<Map<String, TemplateItem>>() {
                });
        templateMsgDTO.setData(data);
        jedisUtil.set(sendRedisKey.toString(), JSON.toJSONString(templateMsgDTO), 604800);

        //异步执行
        Long finalRecordCount = recordCount;
        LOGGER.info("#####自动化模板消息的发送总数是" + finalRecordCount);
        threadPool.execute(() -> {
            try {
                //要发送的json
                TemplateMsg templateMsg = new TemplateMsg();
                templateMsg.setIsLast(false);
                templateMsg.setAppId(request.getAppId());
                templateMsg.setRecordId(templateAutoLog.getId());
                templateMsg.setNewsKey(sendRedisKey.toString());
                LOGGER.info("自动化发送模板消息,:Condition{}", request);

                //分页查询
                int pageSize = 5000;
                long pageCount = (finalRecordCount + pageSize - 1) / pageSize;
                LOGGER.info(String.format("#####自动化发送模板消息的发送总数是：%d 分页是：%d 批次是 %d公众号是：%s", finalRecordCount, pageCount, templateAutoLog.getId(), request.getAppId()));
                String nextOpenid = null;
                boolean isLastPage = false;
                //发送总条数
                for (int pageNum = 1; pageNum <= pageCount; pageNum++) {
                    if (pageNum == pageCount) {
                        isLastPage = true;
                    }
                    try {
                        //分批取出用户信息
                        request.setPage(pageNum);
                        request.setRows(pageSize);
                        List<WxUserInfoSend> userInfoSends = null;
                        // 不同业务查询语句不一样
                        if (TemplateLogTypeEnum.SHOP.getValue() == logType.intValue()) {
                            userInfoSends = wxUserInfoBiz.queryPage(request, nextOpenid);
                        } else if (TemplateLogTypeEnum.GAME.getValue() == logType.intValue()) {
                            userInfoSends = wxUserInfoBiz.queryPage4Game(request, nextOpenid);
                        }

                        //缓解查询压力，每批休眠
                        TimeUnit.SECONDS.sleep(3);

                        if (!CollectionUtils.isEmpty(userInfoSends)) {
                            try {
                                putAutoMsg2MQ(userInfoSends, templateMsg, isLastPage);
                            } catch (Exception e) {
                                LOGGER.error("往kafka消息队列放数据出错:{}", e);
                            }
                        }

                        nextOpenid = userInfoSends.get(userInfoSends.size() - 1).getOpenid();
                    } catch (Exception e) {
                        LOGGER.error("发送自动化模板消息内部错误,:e{}", e);
                    }
                }
            } catch (Exception e) {
                LOGGER.error("发送自动化模板消息错误,:e{}", e);
            }
        });
    }

    /**
     * 将模板消息推送到kafka队列
     *
     * @param userInfoSends
     * @param templateMsg
     * @param isLastPage
     */
    private void putAutoMsg2MQ(List<WxUserInfoSend> userInfoSends, TemplateMsg templateMsg, boolean isLastPage) {
        TemplateMsg templateMsg_ = new TemplateMsg();
        BeanUtils.copyProperties(templateMsg, templateMsg_);
        templateMsg_.setIsLast(isLastPage);

        // 按200条分割一份 扔队列
        List<List<WxUserInfoSend>> partition = Lists.partition(userInfoSends, MQ_LENGTH_GENERALLY);
        partition.stream().forEach(infoDTOS -> {
            List<MsgUserDTO> msgUserDTOS = Lists.newArrayList();
            infoDTOS.stream().forEach(wxUserInfoSend -> {
                MsgUserDTO msgUserDTO = new MsgUserDTO();
                msgUserDTO.setOpenId(wxUserInfoSend.getOpenid());
                msgUserDTO.setNickName(wxUserInfoSend.getNickname());
                msgUserDTOS.add(msgUserDTO);
            });
            templateMsg_.setUsers(msgUserDTOS);
            kafkaProducer.sendAutoTemplateMsg(templateMsg_);
        });
    }
}
